package spring.cloud.tasks.tasks_manager.tasks_manager;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.KeeperException;
import spring.cloud.tasks.tasks_manager.tasks_manager.listener.*;
import spring.cloud.tasks.tasks_manager.tasks_manager.node.Node;
import spring.cloud.tasks.tasks_manager.tasks_manager.service.*;

import javax.annotation.Resource;

@Slf4j
@Setter
@Getter
public class TaskManager {
    private String zooKeeperClusterKey;
    private String namespace;
    private CuratorFramework curatorFramework;
    public ConnectionStateListener connectionStateListener;
    public TreeCacheService treeCacheService;
    public TaskListenersService taskListenersService;
    //
    public AllotTaskItemService allotTaskItemService;
    //
    private ExecutorCleanService executorCleanService;
    @Resource
    private ReportAlarmService reportAlarmService;
    private UpdateTaskConfigService updateJobConfigService;//以后需要注入

    public TaskManager(CuratorFramework curatorFramework, String namespace, String hostValue) {
        this.namespace = namespace;
        this.curatorFramework = curatorFramework;
        //这个启动失败影响最小
        this.executorCleanService = new ExecutorCleanService(curatorFramework, updateJobConfigService);
        //
        this.treeCacheService = new TreeCacheService(namespace, curatorFramework);
        this.taskListenersService = new TaskListenersService(namespace, curatorFramework, treeCacheService, allotTaskItemService);
        this.allotTaskItemService = new AllotTaskItemService(curatorFramework, hostValue, reportAlarmService);
    }

    private void createNodePathIfNotExists(String path) throws Exception {
        if (curatorFramework.checkExists().forPath(path) == null) {
            try {
                curatorFramework.create().creatingParentsIfNeeded().forPath(path);
            } catch (KeeperException.NodeExistsException e) {// NOSONAR
                log.info("node {} already existed, so skip creation", path);
            }
        }
    }

    /**
     * leadership election, add listeners
     */
    public void start() throws Exception {
        start0();
        addConnectionLostListener();
    }

    public void start0() throws Exception {
        treeCacheService.start();
        // create ephemeral node $SaturnExecutors/leader/host & $Jobs.
        allotTaskItemService.leaderElection();
        addLeaderElectionListener();
        //
        taskListenersService.addExistTaskListener();
        addTaskAddOrRemoveListener();
        addExecutorOnlineOfflineAndTrafficListener();
        addExecutorsAllotListener();


    }

    /**
     * watch 1-level-depth of the children of /$SaturnExecutors/leader
     */
    private void addLeaderElectionListener() throws Exception {
        String path = Node.LEADERNODE_PATH;
        int depth = 1;
        createNodePathIfNotExists(path);
        treeCacheService.addTreeCacheIfAbsent(path, depth);
        treeCacheService.addTreeCacheListenerIfAbsent(path, depth, new LeadershipElectionListener(allotTaskItemService));
    }

    /**
     * watch 1-level-depth of the children of /$Jobs
     */
    private void addTaskAddOrRemoveListener() throws Exception {
        String path = Node.TASKS_NODE_PATH;
        int depth = 1;
        createNodePathIfNotExists(path);
        treeCacheService.addTreeCacheIfAbsent(path, depth);
        treeCacheService.addTreeCacheListenerIfAbsent(path, depth, new TaskAddOrRemoveListener(taskListenersService));
    }

    /**
     * watch 2-level-depth of the children of /$SaturnExecutors/executors
     */
    private void addExecutorOnlineOfflineAndTrafficListener() throws Exception {
        String path = Node.EXECUTORSNODE_PATH;
        int depth = 2;
        createNodePathIfNotExists(path);
        treeCacheService.addTreeCacheIfAbsent(path, depth);
        treeCacheService.addTreeCacheListenerIfAbsent(path, depth, new ExecutorOnlineOfflineListener(allotTaskItemService, executorCleanService));
        treeCacheService.addTreeCacheListenerIfAbsent(path, depth, new ExecutorLostOrRecoverTrafficListener(allotTaskItemService));
    }

    /**
     * watch 1-level-depth of the children of /$SaturnExecutors/sharding
     */
    private void addExecutorsAllotListener() throws Exception {
        String path = Node.SHARDINGNODE_PATH;
        int depth = 1;
        createNodePathIfNotExists(path);
        treeCacheService.addTreeCacheIfAbsent(path, depth);
        treeCacheService.addTreeCacheListenerIfAbsent(path, depth, new TasksFullAllotListener(allotTaskItemService));
    }

    private void addConnectionLostListener() {
        connectionStateListener = new ConnectionStateListener(this, "connectionListener-for-NamespaceSharding-" + namespace);
        curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
    }

    //
    public void stopWithCurator() {
        stop();
        curatorFramework.close();
    }

    /**
     * close listeners, delete leadership
     */
    public void stop() {
        try {
            if (connectionStateListener != null) {
                curatorFramework.getConnectionStateListenable().removeListener(connectionStateListener);
                connectionStateListener.shutdownNowUntilTerminated();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        try {
            treeCacheService.shutdown();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        try {
            allotTaskItemService.shutdown();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


}
